package com.lambert.ginkgo;

import cn.hutool.core.util.IdUtil;
import com.lambert.ginkgo.exception.WebClientException;
import com.lambert.ginkgo.properties.WebClientProperties;
import io.netty.channel.ChannelOption;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.codec.DecodingException;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.ClientRequest;
import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.netty.http.client.HttpClient;
import reactor.netty.resources.ConnectionProvider;
import reactor.util.context.Context;
import reactor.util.retry.Retry;

import javax.net.ssl.SSLException;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * WebClient封装类，用于替代传统的RestTemplate
 * 支持响应式编程模型，提供更高效的异步非阻塞HTTP通信
 * 包含完整的日志、追踪、异常处理、重试机制和连接管理功能
 */
@Slf4j
@Service
public class WebClientTemplate {

    // WebClient实例，用于执行HTTP请求
    private final WebClient webClient;

    // 配置属性对象，包含所有可配置参数
    private final WebClientProperties properties;

    // 自定义线程池，用于执行WebClient的订阅操作
    private final ExecutorService executorService;

    /**
     * 构造方法，初始化WebClientTemplate
     * @param properties 配置属性对象，包含所有WebClient相关配置
     */
    public WebClientTemplate(WebClientProperties properties) throws SSLException {
        this.properties = properties;
        // 创建自定义线程池
        this.executorService = createExecutorService();
        // 创建并配置WebClient实例
        this.webClient = createWebClient();
    }

    /**
     * 创建自定义线程池，用于WebClient的订阅操作
     * 线程池参数从配置中获取，确保资源可控
     * @return 配置好的ExecutorService实例
     */
    private ExecutorService createExecutorService() {
        // 创建固定大小的线程池
        return Executors.newFixedThreadPool(
                properties.getThreadPoolSize(),
                // 自定义线程工厂，设置线程名称和守护线程属性
                runnable -> {
                    Thread thread = new Thread(runnable);
                    // 设置线程名称，格式为：前缀+线程ID，便于日志追踪
                    thread.setName(properties.getThreadPoolNamePrefix() + thread.getId());
                    // 将线程设置为守护线程，避免应用关闭时的线程残留
                    thread.setDaemon(true);
                    return thread;
                }
        );
    }

    /**
     * 创建并配置WebClient实例
     * 包含连接池、超时设置、默认头信息和各种过滤器配置
     * @return 配置完成的WebClient实例
     */
    private WebClient createWebClient() throws SSLException {
        // 配置连接池（如果启用）
        ConnectionProvider connectionProvider = null;
        if (properties.isUseConnectionPool()) {
            // 使用ConnectionProvider构建连接池
            connectionProvider = ConnectionProvider.builder("web-client-pool")
                    // 最大连接数
                    .maxConnections(properties.getMaxConnections())
                    // 获取连接的超时时间
                    .pendingAcquireTimeout(Duration.ofMillis(properties.getAcquireTimeout()))
                    // 连接的最大空闲时间
                    .maxIdleTime(Duration.ofMillis(properties.getIdleTimeout()))
                    .build();
        }
        //创建客户端httpClient
        HttpClient httpClient = null;
        if (connectionProvider == null) {
            httpClient = HttpClient.create();
        } else {
            httpClient = HttpClient.create(connectionProvider);
        }
        SslContext sslContext = SslContextBuilder.forClient()
                // 信任所有证书
                .trustManager(InsecureTrustManagerFactory.INSTANCE)
                .build();
        // 配置HTTP客户端（基于Reactor Netty）
        httpClient = httpClient
                .secure(sslContextSpec -> sslContextSpec.sslContext(sslContext))
                // 压缩支持
                .compress(true)
                // 保持连接
                .keepAlive(true)
                // 设置连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, properties.getConnectTimeout())
                // 配置连接建立后的处理器
                .doOnConnected(conn -> conn
                        // 添加读取超时处理器
                        .addHandlerLast(new ReadTimeoutHandler(properties.getReadTimeout(), TimeUnit.MILLISECONDS))
                        // 添加写入超时处理器
                        .addHandlerLast(new WriteTimeoutHandler(properties.getWriteTimeout(), TimeUnit.MILLISECONDS)))
                // 设置响应超时时间
                .responseTimeout(Duration.ofMillis(properties.getResponseTimeout()));

        // 创建WebClient构建器
        WebClient.Builder builder = WebClient.builder()
                // 设置HTTP客户端连接器
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                // 配置默认请求头
                .defaultHeaders(headers -> {
                    // 添加配置中定义的默认请求头
                    properties.getDefaultHeaders().forEach(headers::set);
                });

        // 如果配置了基础URL，则设置
        if (properties.getBaseUrl() != null && !properties.getBaseUrl().isEmpty()) {
            builder.baseUrl(properties.getBaseUrl());
        }

        // 添加各种过滤器：日志、追踪、错误处理等
                // 请求日志过滤器
        builder
                .filter(logRequest())
                // 响应日志过滤器
                .filter(logResponse())
                // TraceId传递过滤器
                .filter(traceIdFilter())
                // 错误处理过滤器
                .filter(errorHandlingFilter());

        // 构建并返回WebClient实例
        return builder.build();
    }

    /**
     * 创建请求日志过滤器
     * 用于记录发出的HTTP请求信息，包括方法、URL、头信息和可选的请求体
     * @return 配置好的请求日志ExchangeFilterFunction
     */
    private ExchangeFilterFunction logRequest() {
        return ExchangeFilterFunction.ofRequestProcessor(request -> {
            // 仅在DEBUG级别日志启用时记录
            if (log.isDebugEnabled()) {
                // 记录请求方法和URL
                log.debug("发送请求: {} {}", request.method(), request.url());
                // 记录请求头信息
                HttpHeaders headers = request.headers();
                log.debug("请求头: {}", headers);
            }
            // 返回原始请求（不做修改）
            return Mono.just(request);
        });
    }

    /**
     * 创建响应日志过滤器
     * 用于记录收到的HTTP响应信息，包括状态码、头信息和可选的响应体
     * @return 配置好的响应日志ExchangeFilterFunction
     */
    private ExchangeFilterFunction logResponse() {
        return ExchangeFilterFunction.ofResponseProcessor(originalResponse -> {
            // 仅在DEBUG级别日志启用时记录
            if (log.isDebugEnabled()) {
                // 记录响应状态码
                log.debug("响应状态码: {}", originalResponse.statusCode());
                // 记录响应头信息
                HttpHeaders httpHeaders = originalResponse.headers().asHttpHeaders();
                log.debug("响应头: {}", httpHeaders);
            }
            // 返回原始响应（不做修改）
            return Mono.just(originalResponse);
        });
    }
    /**
     * 创建TraceId过滤器
     * 用于在分布式系统中传递追踪ID，实现请求的全链路追踪
     * @return 配置好的TraceId ExchangeFilterFunction
     */
    private ExchangeFilterFunction traceIdFilter() {
        return (request, next) -> {
            // 从上下文获取TraceId，如果不存在则生成新的
            return Mono.deferContextual(contextView -> {
                // 尝试从上下文获取TRACE_ID，如果没有则生成新的
                String traceId = contextView.getOrDefault("TRACE_ID", generateTraceId());
                // 创建新的请求，添加TraceId头信息
                ClientRequest newRequest = ClientRequest.from(request)
                        .header(properties.getTraceIdHeaderName(), traceId)
                        .build();
                // 将TraceId存入上下文，以便后续操作使用
                return next.exchange(newRequest)
                        .contextWrite(Context.of("TRACE_ID", traceId));
            });
        };
    }

    /**
     * 创建错误处理过滤器
     * 用于统一处理HTTP错误响应，将其转换为自定义异常
     * @return 配置好的错误处理ExchangeFilterFunction
     */
    private ExchangeFilterFunction errorHandlingFilter() {
        return ExchangeFilterFunction.ofResponseProcessor(response -> {
            // 检查响应状态码是否为错误状态（4xx或5xx）
            if (response.statusCode().isError()) {
                // 读取响应体内容（如果有）
                return response.bodyToMono(String.class)
                        .defaultIfEmpty("") // 如果响应体为空，使用空字符串
                        .flatMap(body -> {
                            // 记录错误信息
                            log.error("请求错误: {}，响应体: {}", response.statusCode(), body);
                            // 抛出自定义异常，包含状态码、响应体和是否可重试的信息
                            return Mono.error(new WebClientException(
                                    "HTTP request failed with status code: " + response.statusCode(),
                                    response.statusCode(),
                                    body,
                                    isRetryableStatusCode(response.statusCode().value())
                            ));
                        });
            }
            // 如果响应正常，直接返回
            return Mono.just(response);
        });
    }

    /**
     * 判断HTTP状态码是否表示可重试的错误
     * 通常服务器错误（5xx）和请求过多（429）是可重试的
     * @param statusCode HTTP响应状态码
     * @return 如果可重试返回true，否则返回false
     */
    private boolean isRetryableStatusCode(int statusCode) {
        // 5xx服务器错误和429请求过多通常是可重试的
        return statusCode >= 500 || statusCode == 429;
    }

    /**
     * 生成默认的TraceId
     * 使用UUID随机生成，确保唯一性
     * @return 生成的TraceId字符串
     */
    private String generateTraceId() {
        return IdUtil.fastSimpleUUID();
    }

    /**
     * 执行普通GET请求
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param responseType 响应数据的类型
     * @param headers 请求头信息
     * @param <T> 响应数据的泛型类型
     * @return 包含响应数据的Mono对象
     */
    public <T> Mono<T> get(String url, Class<T> responseType, HttpHeaders headers) {
        return webClient.get()
                .uri(url)
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                // 执行请求并检索响应
                .retrieve()
                // 将响应体转换为指定类型的Mono
                .bodyToMono(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("GET {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("GET {} 响应体解码失败: {}", url, e.getMessage());
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("GET {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                // 应用重试策略
                .transform(withRetry())
                // 指定订阅线程池
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }

    /**
     * 执行带查询参数的GET请求
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param params 查询参数键值对
     * @param responseType 响应数据的类型
     * @param headers 请求头信息
     * @param <T> 响应数据的泛型类型
     * @return 包含响应数据的Mono对象
     */
    public <T> Mono<T> get(String url, Map<String, Object> params, Class<T> responseType, HttpHeaders headers) {
        return webClient.get()
                .uri(uriBuilder -> {
                    // 构建URI，添加路径和查询参数
                    uriBuilder.path(url);
                    if (params != null) {
                        params.forEach(uriBuilder::queryParam);
                    }
                    return uriBuilder.build();
                })
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                .retrieve()
                .bodyToMono(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("GET {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("GET {} 响应体解码失败: ",url , e);
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("GET {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                .transform(withRetry())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }

    /**
     * 执行普通POST请求
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param requestBody 请求体对象
     * @param responseType 响应数据的类型
     * @param <T> 请求体数据的泛型类型
     * @param <R> 响应数据的泛型类型
     * @return 包含响应数据的Mono对象
     */
    public <T, R> Mono<R> post(String url, T requestBody, Class<R> responseType) {
        return post(url, requestBody, responseType, HttpHeaders.EMPTY);
    }

    /**
     * 执行普通POST请求
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param requestBody 请求体对象
     * @param responseType 响应数据的类型
     * @param headers 请求头信息
     * @param <T> 请求体数据的泛型类型
     * @param <R> 响应数据的泛型类型
     * @return 包含响应数据的Mono对象
     */
    public <T, R> Mono<R> post(String url, T requestBody, Class<R> responseType, HttpHeaders headers) {
        return webClient.post()
                .uri(url)
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                .body(Mono.just(requestBody), requestBody.getClass())
                .retrieve()
                .bodyToMono(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("POST {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("POST {} 响应体解码失败: ", url, e);
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("POST {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                .transform(withRetry())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }
    /**
     * 执行流式POST请求（以Flux作为请求体）
     * 适用于需要发送大量数据或流式数据的场景
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param requestFlux 流式请求体
     * @param requestType 请求体元素的类型
     * @param responseType 响应数据的类型
     * @param headers 请求头信息
     * @param <T> 请求体元素的泛型类型
     * @param <R> 响应数据的泛型类型
     * @return 包含响应数据的Mono对象
     */
    public <T, R> Flux<R> postFlux(String url, Mono<T> requestFlux, Class<T> requestType, Class<R> responseType, HttpHeaders headers) {
        return webClient.post()
                .uri(url)
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                // 设置流式请求体
                .body(requestFlux, requestType)
                .retrieve()
                .bodyToFlux(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("POST {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("POST {} 响应体解码失败: {}", url, e.getMessage());
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("POST {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                .retryWhen(retrySpec())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }
    /**
     * 执行流式POST请求（以Flux作为请求体）
     * 适用于需要发送大量数据或流式数据的场景
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param requestFlux 流式请求体
     * @param requestType 请求体元素的类型
     * @param responseType 响应数据的类型
     * @param headers 请求头信息
     * @param <T> 请求体元素的泛型类型
     * @param <R> 响应数据的泛型类型
     * @return 包含响应数据的Mono对象
     */
    public <T, R> Flux<R> postFlux(String url, Flux<T> requestFlux, Class<T> requestType, Class<R> responseType, HttpHeaders headers) {
        return webClient.post()
                .uri(url)
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                // 设置流式请求体
                .body(requestFlux, requestType)
                .retrieve()
                .bodyToFlux(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("POST {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("POST {} 响应体解码失败: {}", url, e.getMessage());
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("POST {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                .retryWhen(retrySpec())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }

    /**
     * 执行GET请求，获取流式响应
     * 适用于接收大量数据或持续数据流的场景
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param responseType 响应流中元素的类型
     * @param headers 请求头信息
     * @param <T> 响应流元素的泛型类型
     * @return 包含流式响应数据的Flux对象
     */
    public <T> Flux<T> getFlux(String url, Class<T> responseType, HttpHeaders headers) {
        return webClient.get()
                .uri(url)
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                .retrieve()
                // 将响应体转换为指定类型的Flux
                .bodyToFlux(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("POST {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("POST {} 响应体解码失败: {}", url, e.getMessage());
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("POST {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                // 应用重试策略
                .retryWhen(retrySpec())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }

    /**
     * 执行带查询参数的GET请求，获取流式响应
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param params 查询参数键值对
     * @param responseType 响应流中元素的类型
     * @param headers 请求头信息
     * @param <T> 响应流元素的泛型类型
     * @return 包含流式响应数据的Flux对象
     */
    public <T> Flux<T> getFlux(String url, Map<String, Object> params, Class<T> responseType, HttpHeaders headers) {
        return webClient.get()
                .uri(uriBuilder -> {
                    uriBuilder.path(url);
                    if (params != null) {
                        params.forEach(uriBuilder::queryParam);
                    }
                    return uriBuilder.build();
                })
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                .retrieve()
                .bodyToFlux(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("POST {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("POST {} 响应体解码失败: {}", url, e.getMessage());
                    return Mono.justOrEmpty((T) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("POST {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                .retryWhen(retrySpec())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }

    /**
     * 执行流式POST请求，并将全部流式响应组装成Map返回
     * 适用于需要发送流数据并接收流响应，最后聚合结果的场景
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param requestFlux 流式请求体
     * @param requestType 请求体元素的类型
     * @param responseType 响应流中元素的类型
     * @param keyExtractor 从响应元素中提取Map键的函数
     * @param headers 请求头信息
     * @param <T> 请求体元素的泛型类型
     * @param <R> 响应流元素的泛型类型
     * @return 包含聚合结果的Map的Mono对象
     */
    public <T, R> Mono<Map<String, R>> postFluxAndCollect(String url, Flux<T> requestFlux,
                                                          Class<T> requestType, Class<R> responseType,
                                                          Function<R, String> keyExtractor, HttpHeaders headers) {
        return webClient.post()
                .uri(url)
                .body(requestFlux, requestType)
                .retrieve()
                .bodyToFlux(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("POST {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("POST {} 响应体解码失败: {}", url, e.getMessage());
                    return Mono.justOrEmpty((R) getDefaultInstance(responseType));
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("POST {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                // 将流式响应收集到Map中
                .collectMap(keyExtractor)
                .transform(withRetry())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }

    /**
     * 通用请求方法，支持任意HTTP方法和复杂的响应类型
     * 适用于处理特殊请求或复杂的泛型响应类型
     * @param url 请求URL（可以是相对路径或绝对路径）
     * @param method HTTP请求方法（GET, POST, PUT, DELETE等）
     * @param requestBody 请求体对象（可为null）
     * @param responseType 响应数据的类型引用（支持泛型）
     * @param headers 请求头信息
     * @param <T> 响应数据的泛型类型
     * @return 包含响应数据的Mono对象
     */
    public <T> Mono<T> exchange(String url, HttpMethod method, Object requestBody,
                                ParameterizedTypeReference<T> responseType, HttpHeaders headers) {
        return webClient.method(method)
                .uri(url)
                .headers(httpHeaders -> httpHeaders.addAll(headers))
                // 设置请求体（支持null）
                .bodyValue(requestBody)
                .retrieve()
                // 使用ParameterizedTypeReference处理泛型
                .bodyToMono(responseType)
                // 3. 处理空响应体（仅当流正常完成且无数据时触发）
                .switchIfEmpty(Mono.defer(() -> {
                    log.warn("POST {} 返回空响应体", url);
                    // 返回默认值或空对象（避免后续处理NPE）
                    return Mono.justOrEmpty((T) null);
                }))
                // 4. 处理解码错误（如响应体非JSON格式）
                .onErrorResume(DecodingException.class, e -> {
                    log.error("POST {} 响应体解码失败: {}", url, e.getMessage());
                    return Mono.justOrEmpty((T) null);
                })
                // 5. 处理其他网络错误（如连接超时）
                .onErrorResume(e -> {
                    log.error("POST {} 调用失败: {}", url, e.getMessage());
                    return Mono.error(new RuntimeException("接口调用失败: " + e.getMessage()));
                })
                .transform(withRetry())
                .subscribeOn(Schedulers.fromExecutorService(executorService));
    }

    /**
     * 创建重试转换函数，用于Mono类型的响应
     * 将重试策略应用于Mono序列
     * @param <T> 响应数据的泛型类型
     * @return 包含重试逻辑的函数
     */
    private <T> Function<Mono<T>, Mono<T>> withRetry() {
        return mono -> mono.retryWhen(retrySpec());
    }

    /**
     * 配置重试策略
     * 基于指数退避策略，并可配置重试次数、初始退避时间和抖动因子
     * @return 配置好的Retry对象
     */
    private Retry retrySpec() {
        return Retry.backoff(
                        // 最大重试次数
                        properties.getRetryMaxAttempts(),
                        // 初始退避时间
                        Duration.ofMillis(properties.getRetryInitialBackoff()))
                // 抖动因子，避免请求峰值
                .jitter(properties.getRetryJitter())
                .filter(throwable -> {
                    // 只对特定异常进行重试
                    if (throwable instanceof WebClientException) {
                        // 检查异常是否标记为可重试
                        return ((WebClientException) throwable).isRetryable();
                    }
                    // 对IO异常进行重试（如网络错误）
                    return throwable instanceof java.io.IOException;
                })
                .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) ->
                        // 当重试耗尽时，抛出包含原始异常的新异常
                        new WebClientException("重试次数耗尽", retrySignal.failure()));
    }
    /**
     * 创建目标类型的默认实例（避免空指针）
     */
    private <R> R getDefaultInstance(Class<R> type) {
        try {
            // 尝试通过无参构造函数创建实例
            return type.getDeclaredConstructor().newInstance();
        } catch (Exception e) {
            // 无法创建实例时返回null（上游需处理null）
            return null;
        }
    }
    /**
     * 销毁方法，用于释放资源
     * 主要负责关闭自定义线程池，避免资源泄漏
     * 建议在Spring Bean的销毁阶段调用
     */
    @PreDestroy
    public void destroy() {
        if (executorService != null && !executorService.isTerminated()) {
            // 尝试优雅关闭线程池
            executorService.shutdown();
            try {
                // 等待线程池关闭，超时则强制关闭
                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                // 中断时强制关闭
                executorService.shutdownNow();
            }
        }
    }
}